package cn.itcast.flink.transformation;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FilterDemo {
    public static void main(String[] args) throws Exception {
        /**
         * 获取ExecutionEnvironment运行环境
         * 使用fromCollection构建数据源
         * 使用filter操作执行过滤
         * 打印测试
         */
        //TODO 获取ExecutionEnvironment运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 使用fromCollection构建数据源
        DataStream<String> lines = env.readTextFile("./data/input/apache.log");

        //TODO 使用filter操作执行过滤(66.249.73.135)
        SingleOutputStreamOperator<String> result = lines.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String line) throws Exception {
                return line.contains("66.249.73.135");
            }
        });
        //TODO 打印测试
        result.print();

        env.execute();
    }
}